package se.sics.drsm.components;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import se.sics.drsm.bootstrap.client.DRSMBootstrapClient;
import se.sics.drsm.bootstrap.events.DRSMBootstrapCompleted;
import se.sics.drsm.bootstrap.events.DRSMBootstrapResponse;
import se.sics.drsm.bootstrap.port.DRSMBootstrapPort;
import se.sics.drsm.fd.component.PingFailureDetector;
import se.sics.drsm.fd.ports.FailureDetector;
import se.sics.drsm.inits.ChordInit;
import se.sics.drsm.inits.NodeInit;
import se.sics.drsm.model.DRSMAddress;
import se.sics.drsm.model.DRSMConfiguration;
import se.sics.drsm.overlay_address.DRSMOverlayAddress;
import se.sics.drsm.overlay_address.key.NumericRingKey;
import se.sics.drsm.port.drsm_bootstrap.DRSMPeerEntry;
import se.sics.drsm.ports.drsm.DRSMStructuredOverlay;
import se.sics.drsm.ports.drsm.NodePort;
import se.sics.drsm.ports.drsm.events.*;
import se.sics.kompics.*;
import se.sics.kompics.address.Address;
import se.sics.kompics.network.Network;
import se.sics.kompics.p2p.bootstrap.BootstrapConfiguration;
import se.sics.kompics.p2p.bootstrap.BootstrapRequest;
import se.sics.kompics.p2p.bootstrap.client.BootstrapClientInit;
import se.sics.kompics.p2p.fd.ping.PingFailureDetectorInit;
import se.sics.kompics.p2p.monitor.chord.server.ChordMonitorConfiguration;
import se.sics.kompics.p2p.overlay.chord.ChordNeighborsRequest;
import se.sics.kompics.p2p.overlay.chord.ChordNeighborsResponse;
import se.sics.kompics.p2p.overlay.chord.ChordStatus;
import se.sics.kompics.p2p.overlay.chord.LeaveRing;
import se.sics.kompics.timer.Timer;
import se.sics.kompics.web.Web;

import java.util.LinkedList;
import java.util.Set;

public class DRSMNode extends ComponentDefinition {

    // other components
    private Component failureDetector;

    // interfacing components
    private Component bootstrapClient; // monitorClient; //webapp;

    // inner components
    private Component drsm;

    // ports
    Negative<NodePort> nodePort = negative(NodePort.class);
    Positive<Network> networkPort = positive(Network.class);
    Positive<Timer> timerPort = positive(Timer.class);
    Negative<Web> web = negative(Web.class);

    // other instance variables
    private Address self;
    private DRSMAddress drsmSelf;

    private int bootstrapRequestPeerCount;

    private BootstrapConfiguration bootstrapConfiguration;
    private ChordMonitorConfiguration monitorConfiguration;
    private DRSMConfiguration drsmConfiguration;
    private static boolean bootstrapped; // TODO: verify is this thread safe?

    private static Logger logger = LoggerFactory.getLogger(DRSMNode.class);

    public DRSMNode() {
        drsm = create(DistributedRSM.class);
        failureDetector = create(PingFailureDetector.class);
        bootstrapClient = create(DRSMBootstrapClient.class);
        // monitorClient = create(ChordMonitorClient.class);
        // webapp = create(ChordWebApplication.class);

        connect(networkPort, drsm.getNegative(Network.class));
        connect(networkPort, failureDetector.getNegative(Network.class));
        connect(networkPort, bootstrapClient.getNegative(Network.class));
        // connect(networkPort, monitorClient.getNegative(Network.class));
        connect(timerPort, drsm.getNegative(Timer.class));
        connect(timerPort, failureDetector.getNegative(Timer.class));
        connect(timerPort, bootstrapClient.getNegative(Timer.class));
        // connect(timerPort, monitorClient.getNegative(Timer.class));
        // connect(web, webapp.getPositive(Web.class));
        connect(drsm.getNegative(FailureDetector.class),
                failureDetector.getPositive(FailureDetector.class));

        // connect(drsm.getPositive(ChordStatus.class), webapp
        // .getNegative(ChordStatus.class));
        // connect(drsm.getPositive(DRSMStructuredOverlay.class), webapp
        // .getNegative(DRSMStructuredOverlay.class));
        //
        // connect(failureDetector.getPositive(FailureDetectorStatus.class),
        // webapp
        // .getNegative(FailureDetectorStatus.class));
        //

        // connect(drsm.getPositive(ChordStatus.class),
        // monitorClient.getNegative(ChordStatus.class));

        subscribe(handleInit, control);
        subscribe(handleJoinRing, nodePort);
        subscribe(handleJoinRingCompleted,
                drsm.getPositive(DRSMStructuredOverlay.class));
        subscribe(handleBootstrapResponse,
                bootstrapClient.getPositive(DRSMBootstrapPort.class));
        subscribe(handleLeave, nodePort);
        subscribe(handleLeaveDone,
                drsm.getPositive(DRSMStructuredOverlay.class));
        subscribe(handleLookupRequest, nodePort);
        subscribe(handleLookupResponse,
                drsm.getPositive(DRSMStructuredOverlay.class));
        subscribe(handleNeighborsRequest, nodePort);
        subscribe(handleNeighborsResponse, drsm.getPositive(ChordStatus.class));
    }

    Handler<NodeInit> handleInit = new Handler<NodeInit>() {
        public void handle(NodeInit init) {

            self = init.getSelf();
            bootstrapRequestPeerCount = init.getSonConfiguration()
                    .getBootstrapRequestPeerCount();
            bootstrapConfiguration = init.getBootstrapConfiguration();
            monitorConfiguration = init.getMonitorConfiguration();

            // TODO: Implement your own chord component.
            trigger(new ChordInit(init.getSonConfiguration()
                    .getLog2RingSize(), init.getSonConfiguration()
                    .getSuccessorListLength(), init.getSonConfiguration()
                    .getSuccessorStabilizationPeriod(), init
                    .getSonConfiguration().getFingerStabilizationPeriod(),
                    init.getSonConfiguration().getRpcTimeout()),
                    drsm.getControl());
            trigger(new PingFailureDetectorInit(self,
                    init.getFailureDetectorConfiguration()),
                    failureDetector.getControl());
            trigger(new BootstrapClientInit(self,
                    init.getBootstrapConfiguration()),
                    bootstrapClient.getControl());
            // trigger(new
            // ChordMonitorClientInit(init.getMonitorConfiguration(),
            // self), monitorClient.getControl());

            logger.debug(this.getClass().getSimpleName()
                    + " component initialized.");
        }
    };

    Handler<DRSMJoinRing> handleJoinRing = new Handler<DRSMJoinRing>() {
        public void handle(DRSMJoinRing event) {
            // This is the entry event for a node to join the system.
            // A node might end up creating a new node or join an existing
            // group of a node
            logger.debug("{} Handling join an existing group request/create ring.", self);
            drsmSelf = new DRSMAddress(self, event.getNodeKey());

            // TODO: Address format changed
            // trigger(new ChordWebApplicationInit(drsmSelf,
            // monitorConfiguration.getMonitorServerAddress(),
            // bootstrapConfiguration
            // .getBootstrapServerAddress(), monitorConfiguration
            // .getClientWebPort()), webapp.getControl());

            BootstrapRequest request = new BootstrapRequest("Chord",
                    bootstrapRequestPeerCount);
            trigger(request,
                    bootstrapClient.getPositive(DRSMBootstrapPort.class));

        }
    };

    Handler<DRSMBootstrapResponse> handleBootstrapResponse = new Handler<DRSMBootstrapResponse>() {
        public void handle(DRSMBootstrapResponse event) {
            logger.debug(
                    "handling DRSMBootstrapResponse succeded {} already bootstrapped {} ",
                    event.bootstrapSucceeded(), bootstrapped);

            Set<DRSMPeerEntry> somePeers = event.getPeers();
            updateKey(event.getNumericRingKey());

            if (!bootstrapped && somePeers.size() < 1) { // TODO: Discuss what
                // happens if all
                // node dies
                logger.info("triggering DRSMCreateRing  " + drsmSelf);
                trigger(new DRSMCreateRing(drsmSelf),
                        drsm.getPositive(DRSMStructuredOverlay.class));
                bootstrapped = true;
            } else if (!bootstrapped && somePeers.size() > 0) {
                logger.info("triggering DRSMJoinSuccessorGroup  " + drsmSelf);
                triggerJoinRequest(somePeers);
                bootstrapped = true;
            } else {
                // TODO: discuss what is desirable in this scenario.
                logger.error("System in invalid state! Bootstrapped: {} Number of Peers: {} ", bootstrapped, somePeers.size());
            }
        }
    };

    Handler<JoinRingCompleted> handleJoinRingCompleted = new Handler<JoinRingCompleted>() {
        public void handle(JoinRingCompleted event) {
            logger.debug("JoinRing completed");

            DRSMOverlayAddress drsmOverlayAddress = new DRSMOverlayAddress(
                    drsmSelf.getPeerAddress(), drsmSelf.getKey());
            // TODO: bootstrap completed
            // make the the name configurable
            trigger(new DRSMBootstrapCompleted("Chord", drsmOverlayAddress),
                    bootstrapClient.getPositive(DRSMBootstrapPort.class));
        }
    };

    Handler<DRSMLookupRequest> handleLookupRequest = new Handler<DRSMLookupRequest>() {
        public void handle(DRSMLookupRequest event) {
            trigger(event, drsm.getPositive(DRSMStructuredOverlay.class));
        }
    };

    Handler<DRSMLookupResponse> handleLookupResponse = new Handler<DRSMLookupResponse>() {
        public void handle(DRSMLookupResponse event) {
            logger.debug("Received look up reponse {}", event.toString());
            trigger(event, nodePort);
        }
    };

    Handler<LeaveChordRing> handleLeave = new Handler<LeaveChordRing>() {
        public void handle(LeaveChordRing event) {
            trigger(new LeaveRing(),
                    drsm.getPositive(DRSMStructuredOverlay.class));
        }
    };
    Handler<LeaveRingCompleted> handleLeaveDone = new Handler<LeaveRingCompleted>() {
        public void handle(LeaveRingCompleted event) {
            trigger(new LeaveRing(),
                    drsm.getPositive(DRSMStructuredOverlay.class));
        }
    };

    Handler<ChordNeighborsRequest> handleNeighborsRequest = new Handler<ChordNeighborsRequest>() {
        public void handle(ChordNeighborsRequest event) {
            trigger(event, drsm.getPositive(ChordStatus.class));
        }
    };

    Handler<ChordNeighborsResponse> handleNeighborsResponse = new Handler<ChordNeighborsResponse>() {
        public void handle(ChordNeighborsResponse event) {
            trigger(event, nodePort);
        }
    };

    // Helper methods
    public void updateKey(NumericRingKey key) {
        drsmSelf.setKey(key);

    }

    public void triggerJoinRequest(Set<DRSMPeerEntry> somePeers) {
        LinkedList<DRSMAddress> chordInsiders = new LinkedList<DRSMAddress>();
        for (DRSMPeerEntry peerEntry : somePeers) {
            chordInsiders.add(new DRSMAddress(peerEntry.getOverlayAddress()
                    .getPeerAddress(), peerEntry.getOverlayAddress().getKey()));
        }
        trigger(new DRSMJoinSuccessorGroup(drsmSelf, chordInsiders),
                drsm.getPositive(DRSMStructuredOverlay.class));

    }

}